package com.stormragetech.sylwanas.process;

import com.stormragetech.sylwanas.common.RabbitInfo;
import com.stormragetech.sylwanas.constant.Constants;
import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * description: //TODO
 *
 * @author: Xue Bin
 * @version: 1.0.0-SNAPSHOT
 */
@Log4j2
class MessageThreadPoolConsumer {

    private final ConnectionFactory factory;
    private final RabbitInfo info;
    private final RetryCache cache;

    private int threadCount;
    private ExecutorService executor;
    private volatile boolean stop = false;
    private List<MessageConsumer> list;

    MessageThreadPoolConsumer(ConnectionFactory factory, RabbitInfo info, RetryCache cache) {
        this.factory = factory;
        this.info = info;
        this.cache = cache;
        this.threadCount = info.getConsumer().getConsumeThreadCount() * info.getConsumer().getQueue().size();
        this.executor = Executors.newFixedThreadPool(threadCount);
        list = new ArrayList<>();
        info.getConsumer().getQueue().forEach(q -> list.add(new MessageConsumerBuilder(q, this.factory, this.cache, info.getConsumer().getConsumeRetryCount(), info.getInit().isSpring())));
    }

    void start() {
        this.stop = false;
        for (int i = 0; i < threadCount; i++) {
            for (MessageConsumer consumer : list) {
                executor.execute(() -> {
                    while (!stop) {
                        try {
                            if (info.getConsumer().getConsumeIntervalMils() > 0) {
                                TimeUnit.MILLISECONDS.sleep(info.getConsumer().getConsumeIntervalMils());
                            }
                            consumer.consume();
                        } catch (Exception e) {
                            log.error("run exception ", e);
                            e.printStackTrace();
                        }
                    }
                });
            }
        }
    }

    void stop() {
        this.stop = true;
        try {
            TimeUnit.MILLISECONDS.sleep(Constants.ONE_SECOND);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


}
